package org.tao.lightningmq.client.netty;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.tao.lightningmq.client.event.model.TxMessageCallBackEvent;
import org.tao.lightningmq.common.dto.CreateTopicRespDTO;
import org.tao.lightningmq.common.dto.TxMessageCallbackReqDTO;
import org.tao.lightningmq.common.enums.BrokerResponseCode;
import org.tao.lightningmq.common.event.EventBus;
import org.tao.lightningmq.common.remote.SyncFuture;
import org.tao.lightningmq.common.remote.SyncFutureManager;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.Objects;

/**
 * @Author lht
 * @date 2025/1/19 - 17:59
 * @description:
 */
@ChannelHandler.Sharable
public class BrokerRemoteRespHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private static final Logger LOGGER = LoggerFactory.getLogger(BrokerRemoteRespHandler.class);

    private final EventBus eventBus;

    public BrokerRemoteRespHandler(EventBus eventBus) {
        this.eventBus = eventBus;
        eventBus.init();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        if (BrokerResponseCode.TX_CALLBACK_MSG.getCode() == code) {
            TxMessageCallbackReqDTO txMessageCallbackReqDTO = JSON.parseObject(body, TxMessageCallbackReqDTO.class);
            TxMessageCallBackEvent txMessageCallBackEvent = new TxMessageCallBackEvent();
            txMessageCallBackEvent.setMsgId(txMessageCallbackReqDTO.getMessageDTO().getMsgId());
            txMessageCallBackEvent.setContext(channelHandlerContext);
            txMessageCallBackEvent.setTxMessageCallbackReqDTO(txMessageCallbackReqDTO);
            eventBus.publish(txMessageCallBackEvent);
        } else {
            Map map = JSON.parseObject(body, Map.class);
            Object o = map.get("msgId");
            Objects.requireNonNull(o, "msgId is null");
            SyncFuture syncFuture = SyncFutureManager.get((String) o);
            if (syncFuture != null) {
                syncFuture.setResp(tcpMsg);
            }
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
        String reqId = inetSocketAddress.getHostString() + ":" + inetSocketAddress.getPort();
        ctx.attr(AttributeKey.valueOf("reqId")).set(reqId);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("broker connection closed, broker address:{}", ctx.channel().remoteAddress());
        }
        // 触发通道关闭事件
        //BrokerConnectionClosedEvent brokerConnectionClosedEvent = new BrokerConnectionClosedEvent();
        //brokerConnectionClosedEvent.setBrokerReqId((String) ctx.attr(AttributeKey.valueOf("reqId")).get());
        //eventBus.publish(brokerConnectionClosedEvent);
    }
}
